package com.ldzg.blockwatcher.watcher.eth;

import cn.hutool.json.JSONUtil;
import com.ldzg.blockwatcher.config.eth.EthProperties;
import com.ldzg.blockwatcher.constants.EthConstants;
import com.ldzg.blockwatcher.notify.enums.NotifySourceEnum;
import com.ldzg.blockwatcher.risk.RiskEngine;
import com.ldzg.blockwatcher.risk.dto.TransactionInfo;
import com.ldzg.blockwatcher.watcher.AbstractWatcher;
import io.reactivex.disposables.Disposable;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.web3j.abi.EventEncoder;
import org.web3j.abi.TypeReference;
import org.web3j.abi.datatypes.Address;
import org.web3j.abi.datatypes.Event;
import org.web3j.abi.datatypes.Uint;
import org.web3j.protocol.Web3j;
import org.web3j.protocol.core.methods.request.EthFilter;
import org.web3j.protocol.core.methods.response.Log;

import java.math.BigInteger;
import java.util.List;

import static org.web3j.protocol.core.DefaultBlockParameterName.LATEST;

/**
 *  交易预警监听
 *
 * @author: 18310740596@163.com
 * @date: 2025-07-03 23:42:09
 * @version: v1.0
 * @describe: class responsibility
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class EthTradeAlarmWatcher extends AbstractWatcher {

    private final Web3j web3j;
    private final EthProperties ethProperties;
    private final RiskEngine riskEngine;

    private Disposable subscribe;

    @Override
    public void watcher() {
        log.info("以太坊 - 交易预警监听启动");

        Event transferEvent = new Event(EthConstants.TRANSFER_EVENT_NAME,
                                      // 交易来源 from
                              List.of(new TypeReference<Address>(true) {},
                                      // 交易到哪去 to
                                      new TypeReference<Address>(true) {},
                                      new TypeReference<Uint>(true) { }));

        String transferTopic = EventEncoder.encode(transferEvent);

        EthFilter filter = new EthFilter(LATEST,LATEST,ethProperties.getConcatAddress());
        // 添加事件
        filter.addSingleTopic(transferTopic);

        //订阅监听
        this.subscribe = web3j.ethLogFlowable(filter)
                         .subscribe(this::watcherAlarmHandle, ERROR_FUNC);
    }

    private void watcherAlarmHandle(Log watcherLog) {

        log.info("ETH 实时交易信息:{}", JSONUtil.toJsonStr(watcherLog));

        String from = "0x" + watcherLog.getTopics().get(1).substring(26);
        String to = "0x" + watcherLog.getTopics().get(2).substring(26);
        BigInteger value = new BigInteger(watcherLog.getData().substring(2), 16);

        TransactionInfo transactionInfo = TransactionInfo.builder()
                                          .source(NotifySourceEnum.ETH)
                                          .txHash(watcherLog.getTransactionHash())
                                          .fromAddress(from)
                                          .toAddress(to)
                                          .amount(value)
                                          .concatAddress(watcherLog.getAddress())
                                          .raw(watcherLog)
                                          .build();

        riskEngine.check(transactionInfo);
    }

    @Override
    public void dispose() {
        if (this.subscribe != null && !this.subscribe.isDisposed()) {
            this.subscribe.dispose();
        }
    }
}
